package com.geek.framework.study.event.zookeeper;

import com.geek.framework.study.event.EventListener;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.queue.DistributedQueue;
import org.apache.curator.framework.recipes.queue.QueueBuilder;
import org.apache.curator.framework.recipes.queue.QueueConsumer;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @Description zookeeper 分布消息队列发布器（消息以广播的形式发送），使用了{@link DistributedQueue}的能力
 */
public class ZookeeperDistributedEventPublisher extends AbstractDistributedEventPublisher {
    private final ZookeeperMessagingEngineContext context;

    private DistributedQueue queue;

    private final CuratorFramework client;

    private String queueConsumerId;

    private volatile boolean initializedPullFromMessagingEngine = false;

    private volatile boolean initCleanScan = false;


    public ZookeeperDistributedEventPublisher(ZookeeperMessagingEngineContext context) {
        this.context = context;
        client = CuratorFrameworkFactory.newClient(context.getConnectString(), context.getRetryPolicy());
        client.getCuratorListenable().addListener((client, event) -> System.out.println("CuratorEvent: " + event.getType().name()));
        client.start();
    }


    @Override
    protected void afterAddEventListener(EventListener eventListener) {
        if (!initializedPullFromMessagingEngine) {
            try {
                try {
                    client.create().creatingParentContainersIfNeeded().forPath(context.getQueuePath());
                } catch (org.apache.zookeeper.KeeperException.NodeExistsException ignored) {
                    System.out.println(context.getQueuePath() + " is existed, skip......");
                }

                /*
                 * 一个队列发布器，只要添加了业务事件监听，会有一下步骤：
                 * 1. 生成唯一的标识
                 * 2. zookeeper建立对应的元信息路径， queuePath/meta/queueConsumerId
                 * 3. zookeeper建立对应的队列数据路径， queuePath/data/queueConsumerId
                */
                queueConsumerId = UUID.randomUUID().toString();
                // 临时节点，订阅者列表
                client.create()
                        .creatingParentContainersIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .forPath(getQueueMetaPath(queueConsumerId), getQueueDataPath(queueConsumerId).getBytes());
                QueueBuilder<Object> builder = QueueBuilder.builder(
                        client, createQueueConsumer(), context.getQueueSerializer(), getQueueDataPath(queueConsumerId));

                queue = builder.buildQueue();
                queue.start();
                initializedPullFromMessagingEngine = true;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private String getQueueMetaPath(String uuid) {
        return getQueueMetaPathPrefix() + "/" + uuid;
    }

    private String getQueueMetaPathPrefix() {
        return context.getQueuePath() + "/meta";
    }

    private String getQueueDataPath(String uuid) {
        return getQueueDataPathPrefix() + "/" + uuid;
    }

    private String getQueueDataPathPrefix() {
        return context.getQueuePath() + "/data";
    }

    @Override
    protected void pushToMessagingEngine(Object event) throws Exception {
        initCleanScan();
        if (client.checkExists().forPath(getQueueMetaPathPrefix()) == null) {
            System.out.println("no consumer, so drop the event " + event);
            return;
        }

        List<String> pathes = client.getChildren().forPath(getQueueMetaPathPrefix());
        if (pathes.size() == 0) {
            System.out.println("no consumer, so drop the event " + event);
            return;
        }

        /**
         * 1. 扫描queuePath的元信息路径（即queuePath/meta）的子路径，就知道当前的消费者有哪几个
         * 2. 往消费者对应的队列（queuePath/data/queueConsumerId）发送消息
         */
        pathes.forEach(path -> {
            try {
                QueueBuilder<Object> builder = QueueBuilder.builder(client, null, context.getQueueSerializer(),
                        getQueueDataPath(path));
                DistributedQueue queue = builder.buildQueue();
                queue.start();
                queue.put(event);
                CloseableUtils.closeQuietly(queue);
//                System.out.print("push success: " + event + "\n");
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    @Override
    public void doClose() {
        CloseableUtils.closeQuietly(client);
        CloseableUtils.closeQuietly(queue);
    }

    private QueueConsumer<Object> createQueueConsumer() {
        return new QueueConsumer<Object>() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                System.out.println("connection new state: " + newState.name());
            }

            @Override
            public void consumeMessage(Object message) throws Exception {
                receiveFromMessagingEngine(message);
            }
        };
    }


    /**
     * 扫描已经断开的消费者，删除对应的数据节点
     */
    private void initCleanScan() {
        if (!initCleanScan) {
            initCleanScan = true;
            Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
                try {
                    if (client.checkExists().forPath(getQueueDataPathPrefix()) == null) {
                        System.out.println("no consumer, so skip check");
                        return;
                    }
                    List<String> pathes = client.getChildren().forPath(getQueueDataPathPrefix());
                    if (pathes.size() == 0) {
                        System.out.println("no consumer, so skip check");
                        return;
                    }
                    pathes.forEach(path -> {
                        try {
                            // 说明已经断开监听
                            if (client.checkExists().forPath(getQueueMetaPath(path)) == null) {
                                // 删除一个节点，并且递归删除其所有的子节点
                                client.delete().deletingChildrenIfNeeded().forPath(getQueueDataPath(path));
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    });
                } catch (Throwable e) {
                    e.printStackTrace();
                }
            }, 5, 5, TimeUnit.SECONDS);
        }
    }
}
